package com.xl.flinkdemo.streamwc;

import com.xl.flinkdemo.wc.WordCount;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 基于流处理操作的API
 **/
public class StreamWordCount {

  public static void main(String[] args) throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    // 从文件中读取数据

    env.setParallelism(4);

    /*String inputPath = "D:\\studyspace\\flinkdemo\\src\\main\\resources\\";
    DataStream<String> inputdataStream = env.readTextFile(inputPath);*/

    // 用paramter tool工具从程序启动参数中提取配置项
    ParameterTool parameterTool = ParameterTool.fromArgs(args);
    String host =parameterTool.get("hostname");
    int port = parameterTool.getInt("port");



    // 从socket文本流读取数据
    // nc -lk 7777(linux自带的Navicat工具)
    DataStream<String> inputdataStream = env.socketTextStream(host,port);

    // 基于数据流进行转换计算
    SingleOutputStreamOperator<Tuple2<String, Integer>> resultStream =
        inputdataStream.flatMap(new WordCount.MyFlatMapper()).keyBy(0).sum(1);

    //没有数据，这里print只是把任务定义好而已
    resultStream.print();

    //这个方法在流处理中才是真正的启动任务
    // 挨个叠加
    env.execute();
   /* 前面这个数字代表真实执行过程中并行子任务分区，为什么是4呢，开发环境下默认并行度是根据内核来定的
   可以设置：env.setParallelism(4)
    4> (yes,1)
    3> (hello,1)
    1> (gril,1)
    3> (hello,2)
    7> (flink,1)
    4> (are,1)
    8> (ok,1)
    3> (hello,3)
    6> (how,1)
    5> (you,1)
    7> (boy,1)*/
  }
}
